package com.feidee.fd.sml.algorithm.component.feature

import org.apache.spark.ml.PipelineStage
import org.apache.spark.ml.feature.Interaction
import org.apache.spark.sql.DataFrame

/**
  * @Author tangjinyuan
  * @Date 2019/03/21 16:33
  * @Description InteractionEncoder  特征组件
  * @Reviewer Interaction
  */
case class InteractEncoderParam(
                                 override val input_pt: String,
                                 override val output_pt: String,
                                 override val hive_table: String,
                                 override val flow_time: String,
                                 // 需要相乘的向量列，多列用逗号分隔
                                 override val inputCol: String,
                                 override val outputCol: String,
                                 override val preserveCols: String,
                                 override val modelPath: String
                               ) extends FeatureParam {

  def this() = this(null, null, null, null, "input", "features", null, null)

}


class InteractEncoder extends AbstractFeatureEncoder[InteractEncoderParam] {

  override def setUp(param: InteractEncoderParam, data: DataFrame): Array[PipelineStage] = {

    val interaction = new Interaction()
      .setInputCols(param.inputCol.split(","))
      .setOutputCol(param.outputCol)

    Array(interaction)
  }

}

object InteractEncoder {

  def apply(paramStr: String): Unit = {
    new InteractEncoder()(paramStr)
  }

  def main(args: Array[String]): Unit = {
    InteractEncoder(args(0))
  }
}


